package com.surfilter.massdata.spark.task;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import scala.Tuple2;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN071;
import com.surfilter.massdata.spark.bean.SAN081;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.IPDatabase;

@SuppressWarnings("serial")
public class NetizenAnalyzTask implements ISparkTask {
	private String city;
	private String oper;
	private String iptype;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,
			Map<String, String> commandMap) {
		List<OutQueueEntity> list = new ArrayList<OutQueueEntity>();
		try{
			// 读取配置数据
			DataFrame dns = dataFrames.get("cip_frame");
			SQLContext sqlContext = dns.sqlContext();
			String dayStr = commandMap.get("-d");
			Date date = DateUtil.getExecDate(new Date(), dayStr);

			JavaRDD<Map<String, Object>> rdd = getNetizenRDD(dns);
			
			// 网民分布统计-地市网民访问量
			DataFrame city = getCityDF(sqlContext, date,rdd);
			list.add(new OutQueueEntity(this.city, city));

			// 网民分布统计-运营商网民访问量
			DataFrame oper = getOperatorDF(sqlContext, date,rdd);
			list.add(new OutQueueEntity(this.oper, oper));

			// 网民分布统计-网民访问量分类统计
			DataFrame iptype = getIpTypeDF(sqlContext, date,rdd);
			list.add(new OutQueueEntity(this.iptype, iptype));
			//System.out.println("city:"+city.count());
			//System.out.println("oper:"+oper.count());
			//System.out.println("iptype:"+iptype.count());
			CommonUtils.deleteTaskTableData("SAN081", date, "COUNT_AREA_SUM", 0, "day");
			CommonUtils.deleteTaskTableData("SAN081", date, "COUNT_OPERATORS_SUM", 0, "day");
			CommonUtils.deleteTaskTableData("SAN071", date, "COUNT_IP_TYPE", 0, "day");
			
		}catch(Exception e){
			SysLog.error(e.getMessage());
		}
		
		return list;
	}

	@SuppressWarnings("resource")
	private JavaRDD<Map<String, Object>> getNetizenRDD(DataFrame dns) {
		JavaSparkContext sc = new JavaSparkContext(dns.sqlContext().sparkContext());
		final Broadcast<IPDatabase> ipBroadcast = sc.broadcast(IPDatabase.getInstance());
		
		JavaRDD<Map<String,Object>> rdd = dns.toJavaRDD().map(new Function<Row, Map<String, Object>>() {

			@Override
			public Map<String, Object> call(Row row) throws Exception {
				Map<String,Object> map = new HashMap<String,Object>();
				try{
					String cip = row.getAs("cip");
					StructIP structCip = ipBroadcast.value().get(cip);
					if(structCip != null){
						map.put("dns_visit_count",row.getAs("dns_visit_count"));
						map.put("city", structCip.get(DnsStatConsts.IPExField.CITY));
						map.put("operators_id", structCip.get(DnsStatConsts.IPExField.OPERATORS_ID));
						map.put("ip_type", structCip.get(DnsStatConsts.IPExField.IP_TYPE));
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return map;
			}
		});
		return rdd;
	}

	private DataFrame getCityDF(SQLContext sqlContext,final Date date,JavaRDD<Map<String,Object>> rdd) {
		JavaRDD<SAN081> cityRDD = rdd.mapToPair(new PairFunction<Map<String,Object>,String,Long>() {

			@Override
			public Tuple2<String,Long> call(Map<String, Object> map){
				try{
					String city = CommonUtils.valueOf(map.get("city"));
					if(StringUtils.isNotBlank(city)){
						long dns_visit_count = Long.parseLong(CommonUtils.valueOf(map.get("dns_visit_count")));
						return new Tuple2<String,Long>(city,dns_visit_count);
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return new Tuple2<String,Long>("",0L);
			}
		}).reduceByKey(new Function2<Long, Long, Long>() {
			
			@Override
			public Long call(Long v1, Long v2){
				return v1 + v2;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Tuple2<String,Long>>,SAN081>() {

			@Override
			public Iterable<SAN081> call(Iterator<Tuple2<String, Long>> it){
				List<SAN081> list = new ArrayList<SAN081>();
				try{
					while(it.hasNext()){
						Tuple2<String,Long> tuple = it.next();
						String city = tuple._1;
						if(StringUtils.isNotBlank(city)){
							SAN081 sa = new SAN081();
							sa.setYear(DateUtil.getCurrentYear(date));
							sa.setHalf_year(DateUtil.getHalfYear(date));
							sa.setQuarter(DateUtil.getQuarter(date));
							sa.setMonth(DateUtil.getCurrentMonth(date));
							sa.setWeek(DateUtil.getCurrentWeek(date));
							sa.setDay(DateUtil.getCurrentDay(date));
							sa.setBuss_type("COUNT_AREA_SUM");
							sa.setBuss_value(city);
							sa.setCount_value(tuple._2);
							sa.setSta_range(0);
							list.add(sa);
						}
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});

		DataFrame df = sqlContext.createDataFrame(cityRDD, SAN081.class);
		return df;
	}

	private DataFrame getOperatorDF(SQLContext sqlContext, final Date date,JavaRDD<Map<String,Object>> rdd) {
		JavaRDD<SAN081> operatorRDD = rdd.mapToPair(new PairFunction<Map<String,Object>,String,Long>() {

			@Override
			public Tuple2<String,Long> call(Map<String, Object> map){
				try{
					String operators_id = CommonUtils.valueOf(map.get("operators_id"));
					if(StringUtils.isNotBlank(operators_id)){
						long dns_visit_count = Long.parseLong(CommonUtils.valueOf(map.get("dns_visit_count")));
						return new Tuple2<String,Long>(operators_id,dns_visit_count);
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return new Tuple2<String,Long>("",0L);
			}
		}).reduceByKey(new Function2<Long, Long, Long>() {
			
			@Override
			public Long call(Long v1, Long v2){
				return v1 + v2;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Tuple2<String,Long>>,SAN081>() {

			@Override
			public Iterable<SAN081> call(Iterator<Tuple2<String, Long>> it){
				List<SAN081> list = new ArrayList<SAN081>();
				try{
					while(it.hasNext()){
						Tuple2<String,Long> tuple = it.next();
						String operators_id = tuple._1;
						if(StringUtils.isNotBlank(operators_id)){
							SAN081 sa = new SAN081();
							sa.setYear(DateUtil.getCurrentYear(date));
							sa.setHalf_year(DateUtil.getHalfYear(date));
							sa.setQuarter(DateUtil.getQuarter(date));
							sa.setMonth(DateUtil.getCurrentMonth(date));
							sa.setWeek(DateUtil.getCurrentWeek(date));
							sa.setDay(DateUtil.getCurrentDay(date));
							sa.setBuss_type("COUNT_OPERATORS_SUM");
							sa.setBuss_value(operators_id);
							sa.setCount_value(tuple._2);
							sa.setSta_range(0);
							list.add(sa);
						}
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		DataFrame df = sqlContext.createDataFrame(operatorRDD, SAN081.class);
		return df;
	}

	private DataFrame getIpTypeDF(SQLContext sqlContext,final Date date,JavaRDD<Map<String,Object>> rdd) {
		JavaRDD<SAN071> ipTypeRDD = rdd.mapToPair(new PairFunction<Map<String,Object>,String,Long>() {

			@Override
			public Tuple2<String,Long> call(Map<String, Object> map){
				try{
					String ip_type = CommonUtils.valueOf(map.get("ip_type"));
					if(StringUtils.isNotBlank(ip_type)){
						long dns_visit_count = Long.parseLong(CommonUtils.valueOf(map.get("dns_visit_count")));
						return new Tuple2<String,Long>(ip_type,dns_visit_count);
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return new Tuple2<String,Long>("",0L);
			}
		}).reduceByKey(new Function2<Long, Long, Long>() {
			
			@Override
			public Long call(Long v1, Long v2){
				return v1 + v2;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Tuple2<String,Long>>,SAN071>() {

			@Override
			public Iterable<SAN071> call(Iterator<Tuple2<String, Long>> it){
				List<SAN071> list = new ArrayList<SAN071>();
				try{
					while(it.hasNext()){
						Tuple2<String,Long> tuple = it.next();
						String ip_type = tuple._1;
						if(StringUtils.isNotBlank(ip_type)){
							SAN071 sa = new SAN071();
							sa.setYear(DateUtil.getCurrentYear(date));
							sa.setHalf_year(DateUtil.getHalfYear(date));
							sa.setQuarter(DateUtil.getQuarter(date));
							sa.setMonth(DateUtil.getCurrentMonth(date));
							sa.setWeek(DateUtil.getCurrentWeek(date));
							sa.setDay(DateUtil.getCurrentDay(date));
							sa.setBuss_type("COUNT_IP_TYPE");
							sa.setBuss_value(ip_type);
							sa.setCount_value(tuple._2);
							sa.setSta_range(0);
							list.add(sa);
						}
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		DataFrame df = sqlContext.createDataFrame(ipTypeRDD, SAN071.class);
		return df;
	}
}